package com.atguigu.gmall.realtime.app.dwd.db;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.BaseDbTableProcessFunction;
import com.atguigu.gmall.realtime.beans.BaseDbTableProcess;
import com.atguigu.gmall.realtime.utils.KafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * Created by 黄凯 on 2023/7/11 0011 16:40
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 事实表动态分流处理
 * * 对于处理过程比较简单的事实表，在当前类中统一的进行动态分流处理
 * * DWD层
 * *      实现方式
 * *          FlinkAPI
 * *          FlinkSQL
 * *      数据域划分
 * *          流量域
 * *              错误日志事实表、启动日志事实表、页面日志事实表、曝光日志事实表、动作日志事实表
 * *              知识点：侧输出流、状态编程
 * *          交易域
 * *              加购、下单、取消订单、支付成功、退单、退款成功
 * *              知识点：FlinkSQL相关知识（连接器、join）
 * *          互动域
 * *              评论、收藏
 * *          工具域
 * *              优惠券领取、优惠券使用
 * *          用户域
 * *              用户注册
 * *      *** 动态分流开发流程 ***
 * *          基本环境准备
 * *          检查点相关设置
 * *          从topic_db中读取主流业务数据
 * *          类型转化以及ETL   jsonStr->jsonObj
 * *
 * *          使用FlinkCDC读取配置表数据--配置流
 * *          对配置流数据进行广播--广播流
 * *
 * *          将主流和广播流进行关联--connect
 * *          对关联之后的数据进行处理--process
 * *          BaseDbTableProcessFunction extends BroadcastProcessFunction{
 * *              open:将配置信息预加载到程序中
 * *              processElement:处理主流数据
 * *                  根据表名 + 操作类型到广播状态以及configMap中获取对应的配置信息
 * *                  如果配置信息不为空，将需要动态处理的事实表数据发送到下游
 * *                      过滤不需要传递的属性
 * *                      补充输出目的地
 * *                      补充事件时间字段
 * *              processBroadcastElement:处理广播流数据
 * *                  op="d":将配置信息从广播状态以及configMap中删除掉
 * *                  op!="d":将配置信息放到广播状态以及configMap中
 * *          }
 * *      将流中的数据写到kafka的不同的主题中
 * *          在KafkaUtil工具类中，封装getKafkaSinkBySchema方法，获取KafkaSink对象
 * *          在处理流中数据的时候，我们需要自己实现序列化
 */
public class BaseDBApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.基本环境贮备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "topic_db";
        String groupId = "base_db_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);

        //3.3 消费数据 封装为流
        DataStreamSource<String> kafksStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 4.对流中的数据类型进行转换并进行简单的ETL  jsonStr->jsonObj
        /**
         * {"database":"gmall","xid":116618,"data":{"payment_way":"3501","refundable_time":"2022-06-15 16:40:11",
         * "original_total_amount":2899.00,"order_status":"1002","consignee_tel":"13969161791",
         * "trade_body":"小米电视E65X 65英寸 全面屏 4K超高清HDR 蓝牙遥控内置小爱 2+8GB AI人工智能液晶网络平板电视 L65M5-EA等1件商品",
         * "id":475079,"operate_time":"2022-06-08 16:40:12","consignee":"郝祥1","create_time":"2022-06-08 16:40:11",
         * "coupon_reduce_amount":0.00,"out_trade_no":"144414959299497","total_amount":2899.00,"user_id":1235,"province_id":15,"activity_reduce_amount":0.00},
         * "old":{"consignee":"郝祥"},"commit":true,"type":"update","table":"order_info","ts":1689065398}
         */
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafksStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            String type = jsonObj.getString("type");
                            if (!type.startsWith("bootstrap-")) {
                                out.collect(jsonObj);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
        );
//         jsonObjDS.print(">>>>");

        //TODO 5.使用FlinkCDC读取配置表数据
        /**
         * {"before":null,"after":{"source_table":"favor_info","source_type":"insert","sink_table":"dwd_interaction_favor_add",
         * "sink_columns":"id,user_id,sku_id,create_time"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source",
         * "ts_ms":0,"snapshot":"false","db":"gmall0201_config","sequence":null,"table":"table_process_dwd","server_id":0,"gtid":null,"file":"",
         * "pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1689065594057,"transaction":null}
         */
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall0201_config")
                .tableList("gmall0201_config.table_process_dwd")
                .username("root")
                .password("000000")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
        DataStreamSource<String> mysqlDS
                = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql_source");

//         mysqlDS.print(">>>>");

        //TODO 6.将配置信息进行广播-broadcast
        MapStateDescriptor<String, BaseDbTableProcess> mapStateDescriptor
                = new MapStateDescriptor<>("mapStateDescriptor", String.class, BaseDbTableProcess.class);

        BroadcastStream<String> broadcastDS = mysqlDS.broadcast(mapStateDescriptor);

        //TODO 7.将主流业务数据和广播流配置信息进行关联-connect
        BroadcastConnectedStream<JSONObject, String> connectDS = jsonObjDS.connect(broadcastDS);

        //TODO 8.对关联之后的数据进行处理-process
        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(
                new BaseDbTableProcessFunction(mapStateDescriptor)
        );

        //TODO 9.将事实表数据写到kafka的不同主题中
        realDS.print(">>>>");

        DataStreamSink<JSONObject> sinkTable = realDS.sinkTo(
                KafkaUtil.getKafkaSinkBySchema(
                        new KafkaRecordSerializationSchema<JSONObject>() {
                            @Nullable
                            @Override
                            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj,
                                                                            KafkaSinkContext context,
                                                                            Long timestamp) {

                                String topic = jsonObj.getString("sink_table");
                                jsonObj.remove("sink_table");


                                return new ProducerRecord<>(topic, jsonObj.toJSONString().getBytes());


                            }
                        }
                )
        );


        env.execute();

    }

}
